package com.googlecode.perfectable.eventable.routing;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class Sources {
	private static final class SingleEventSignaller<M> implements Sink<M> {
		private M result;
		private final Condition semaphore = new ReentrantLock().newCondition();

		static <M> M waitFor(Source<M> source) throws InterruptedException {
			SingleEventSignaller<M> sink = new SingleEventSignaller<>();
			return sink.await(source);
		}

		@Override
		public void receive(M event) {
			this.result = event;
			this.semaphore.signal();
		}

		public M await(Source<M> source) throws InterruptedException {
			source.listen(this);
			try {
				this.semaphore.await();
			}
			finally {
				source.stop(this);
			}
			return this.result;
		}
	}

	public static <M> M waitFor(Source<M> source) throws InterruptedException {
		return SingleEventSignaller.waitFor(source);
	}

	private static final class EmptySource<M> extends BaseSource<M> {

		static final EmptySource<?> INSTANCE = new EmptySource<>();

		private EmptySource() {
			// singleton
		}

		@SuppressWarnings("unchecked")
		public <X> EmptySource<X> casted() {
			return (EmptySource<X>) this;
		}
	}

	public static <M> Source<M> empty() {
		return EmptySource.INSTANCE.casted();
	}
}
